package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseAppV2;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.bean.PaymentWide;
import com.atguigu.gmall.realtime.bean.ProductStats;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.util.DimAsyncFunction;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import com.atguigu.gmall.realtime.util.MySinkUtil;
import com.atguigu.gmall.realtime.util.MyTimeUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/23 14:17
 */
public class DWSProductStatsApp extends BaseAppV2 {
    public static void main(String[] args) {
        String[] topics = {
            "dwd_page_log",
            "dwd_display_log",
            "dwd_favor_info",
            "dwd_cart_info",
            "dwm_order_wide",
            "dwm_payment_wide",
            "dwd_order_refund_info",
            "dwd_comment_info"
        };
        new DWSProductStatsApp().init(30002, 2, "DWSProductStatsApp", "DWSProductStatsApp", topics);
        
    }
    
    @Override
    public void run(StreamExecutionEnvironment env,
                    Map<String, DataStreamSource<String>> sourceStreams) {
        // 1. 解析8个流, 合并成一个流
        final DataStream<ProductStats> productStatsDataStream = parseStreamAndUnionOneStream(sourceStreams);
        // 2. 开窗聚合
        final SingleOutputStreamOperator<ProductStats> productStatsAgg = aggregateByDim(productStatsDataStream);
        // 3. 读取维度
        final SingleOutputStreamOperator<ProductStats> productStatsStream = joinDims(productStatsAgg);
        
        // 4. 数据写入到ClickHouse中
        sink2ClickHouse(productStatsStream);
        
        // 5. 数据写入到Kafka中, 为商品行为关键字需求准备数据
        sink2Kafka(productStatsStream);
    }
    
    private void sink2Kafka(SingleOutputStreamOperator<ProductStats> productStatsStream) {
//        productStatsStream.print();
        productStatsStream
            .map(JSON::toJSONString)
            .addSink(MyKafkaUtil.getFlinkKafkaSink("dws_product_stats"));
    }
    
    private void sink2ClickHouse(SingleOutputStreamOperator<ProductStats> productStatsStream) {
        productStatsStream.addSink(MySinkUtil.getClickHouseSink("gmall2021", "product_stats_2021", ProductStats.class));
    }
    
    private SingleOutputStreamOperator<ProductStats> joinDims(SingleOutputStreamOperator<ProductStats> productStatsAgg) {
        return AsyncDataStream.unorderedWait(
            productStatsAgg,
            new DimAsyncFunction<ProductStats>() {
                
                @Override
                public void addDim(ProductStats input,
                                   Jedis redisClient) {
                    //先读取维度
                    // 1. 读取sku_info
                    JSONObject skuInfo = readDim(redisClient, "DIM_SKU_INFO", input.getSku_id().toString());
                    input.setSku_name(skuInfo.getString("SKU_NAME"));
                    input.setSku_price(skuInfo.getBigDecimal("PRICE"));
                    input.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));
                    input.setSpu_id(skuInfo.getLong("SPU_ID"));
                    input.setTm_id(skuInfo.getLong("TM_ID"));
                    
                    // 2. 读取spu_info
                    final JSONObject spuInfo = readDim(redisClient, "DIM_SPU_INFO", input.getSpu_id().toString());
                    input.setSpu_name(spuInfo.getString("SPU_NAME"));
                    
                    // 3. 读取品牌表
                    final JSONObject tmInfo = readDim(redisClient, "DIM_BASE_TRADEMARK", input.getTm_id().toString());
                    input.setTm_name(tmInfo.getString("TM_NAME"));
                    
                    // 4. 读取三级品类
                    final JSONObject c3Info = readDim(redisClient, "DIM_BASE_CATEGORY3", input.getCategory3_id().toString());
                    input.setCategory3_name(c3Info.getString("NAME"));
                }
            },
            30,
            TimeUnit.SECONDS
        );
    }
    
    private SingleOutputStreamOperator<ProductStats> aggregateByDim(DataStream<ProductStats> productStatsDataStream) {
        return productStatsDataStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                    .withTimestampAssigner((ps, ts) -> ps.getTs())
            )
            .keyBy(ProductStats::getSku_id)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(
                new ReduceFunction<ProductStats>() {
                    @Override
                    public ProductStats reduce(ProductStats s1,
                                               ProductStats s2) throws Exception {
                        s1.setFavor_ct(s1.getFavor_ct() + s2.getFavor_ct());
                        s1.setClick_ct(s1.getClick_ct() + s2.getClick_ct());
                        s1.setDisplay_ct(s1.getDisplay_ct() + s2.getDisplay_ct());
                        s1.setCart_ct(s1.getCart_ct() + s2.getCart_ct());
                        
                        s1.setOrder_amount(s1.getOrder_amount().add(s2.getOrder_amount()));
                        s1.setOrder_sku_num(s1.getOrder_sku_num() + s2.getOrder_sku_num());
                        
                        s1.setPayment_amount(s1.getPayment_amount().add(s2.getPayment_amount()));
                        
                        s1.setRefund_amount(s1.getRefund_amount().add(s2.getRefund_amount()));
                        
                        s1.setComment_ct(s1.getComment_ct() + s2.getComment_ct());
                        s1.setGood_comment_ct(s1.getGood_comment_ct() + s2.getGood_comment_ct());
                        
                        s1.getOrderIdSet().addAll(s2.getOrderIdSet());
                        s1.getPaidOrderIdSet().addAll(s2.getPaidOrderIdSet());
                        s1.getRefundOrderIdSet().addAll(s2.getRefundOrderIdSet());
                        
                        return s1;
                    }
                }, new ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                    @Override
                    public void process(Long key,
                                        Context ctx,
                                        Iterable<ProductStats> elements,
                                        Collector<ProductStats> out) throws Exception {
                        final ProductStats ps = elements.iterator().next();
                        final TimeWindow w = ctx.window();
                        
                        ps.setStt(MyTimeUtil.toDateTimeString(w.getStart()));
                        ps.setEdt(MyTimeUtil.toDateTimeString(w.getEnd()));
                        
                        ps.setOrder_ct((long) ps.getOrderIdSet().size());
                        ps.setPaid_order_ct((long) ps.getPaidOrderIdSet().size());
                        ps.setRefund_order_ct((long) ps.getRefundOrderIdSet().size());
                        
                        out.collect(ps);
                    }
                });
        
    }
    
    private DataStream<ProductStats> parseStreamAndUnionOneStream(Map<String, DataStreamSource<String>> sourceStreams) {
        // 1. 解析得到页面点击流
        final SingleOutputStreamOperator<ProductStats> productClickStream = sourceStreams
            .get("dwd_page_log")
            .flatMap(new FlatMapFunction<String, ProductStats>() {
                @Override
                public void flatMap(String json,
                                    Collector<ProductStats> out) throws Exception {
                    final JSONObject obj = JSON.parseObject(json);
                    final JSONObject pageObj = obj.getJSONObject("page");
                    final String pageId = pageObj.getString("page_id");
                    if ("good_detail".equalsIgnoreCase(pageId)) {
                        final Long skuId = pageObj.getLong("item");
                        final Long ts = obj.getLong("ts");
                        
                        final ProductStats ps = new ProductStats();
                        ps.setSku_id(skuId);
                        ps.setClick_ct(1L);
                        ps.setTs(ts);
                        out.collect(ps);
                    }
                }
            });
        
        // 2. 曝光率
        final SingleOutputStreamOperator<ProductStats> displayStream = sourceStreams
            .get("dwd_display_log")
            .process(new ProcessFunction<String, ProductStats>() {
                @Override
                public void processElement(String json,
                                           Context ctx,
                                           Collector<ProductStats> out) throws Exception {
                    final JSONObject obj = JSON.parseObject(json);
                    final String itemType = obj.getString("item_type");
                    if ("sku_id".equalsIgnoreCase(itemType)) {
                        final ProductStats ps = new ProductStats();
                        final Long skuId = obj.getLong("item");
                        final Long ts = obj.getLong("ts");
                        ps.setSku_id(skuId);
                        ps.setDisplay_ct(1L);
                        ps.setTs(ts);
                        
                        out.collect(ps);
                    }
                }
            });
        
        // 3.收藏流
        final SingleOutputStreamOperator<ProductStats> favorStream = sourceStreams
            .get("dwd_favor_info")
            .map(json -> {
                final JSONObject obj = JSON.parseObject(json);
                final Long skuId = obj.getLong("sku_id");
                final Long ts = MyTimeUtil.toTs(obj.getString("create_time"));
                
                final ProductStats ps = new ProductStats();
                ps.setSku_id(skuId);
                ps.setTs(ts);
                ps.setFavor_ct(1L);
                return ps;
            });
        // 4. 购物车流
        final SingleOutputStreamOperator<ProductStats> cartStream = sourceStreams
            .get("dwd_cart_info")
            .map(json -> {
                final JSONObject obj = JSON.parseObject(json);
                final Long skuId = obj.getLong("sku_id");
                final Long ts = MyTimeUtil.toTs(obj.getString("create_time"));
                final ProductStats ps = new ProductStats();
                ps.setSku_id(skuId);
                ps.setTs(ts);
                ps.setCart_ct(1L);
                return ps;
            });
        // 5. 订单流
        final SingleOutputStreamOperator<ProductStats> orderStream = sourceStreams
            .get("dwm_order_wide")
            .map(json -> {
                final OrderWide orderWide = JSON.parseObject(json, OrderWide.class);
                
                final ProductStats ps = new ProductStats();
                ps.setSku_id(orderWide.getSku_id());
                ps.setTs(MyTimeUtil.toTs(orderWide.getCreate_time()));
                ps.getOrderIdSet().add(orderWide.getOrder_id());
                System.out.println(ps.getOrderIdSet());
                ps.setOrder_amount(orderWide.getSplit_total_amount());
                ps.setOrder_sku_num(orderWide.getSku_num());
                
                return ps;
            });
        // 6. 支付流
        final SingleOutputStreamOperator<ProductStats> paymentStream = sourceStreams
            .get("dwm_payment_wide")
            .map(json -> {
                final PaymentWide paymentWide = JSON.parseObject(json, PaymentWide.class);
                final ProductStats ps = new ProductStats();
                ps.setSku_id(paymentWide.getSku_id());
                ps.setTs(MyTimeUtil.toTs(paymentWide.getPayment_create_time()));
                ps.getPaidOrderIdSet().add(paymentWide.getOrder_id());
                ps.setPayment_amount(paymentWide.getSplit_total_amount());
                
                return ps;
            });
        
        // 7. 退款流
        final SingleOutputStreamOperator<ProductStats> refundStream = sourceStreams
            .get("dwd_order_refund_info")
            .map(json -> {
                final JSONObject obj = JSON.parseObject(json);
                final Long skuId = obj.getLong("sku_id");
                final Long ts = MyTimeUtil.toTs(obj.getString("create_time"));
                final BigDecimal refundAmount = obj.getBigDecimal("refund_amount");
                
                final ProductStats ps = new ProductStats();
                ps.setSku_id(skuId);
                ps.setTs(ts);
                ps.getRefundOrderIdSet().add(obj.getLong("order_id"));
                ps.setRefund_amount(refundAmount);
                
                return ps;
            });
        
        // 8. 评论流
        final SingleOutputStreamOperator<ProductStats> commentStream = sourceStreams
            .get("dwd_comment_info")
            .map(json -> {
                final JSONObject obj = JSON.parseObject(json);
                final Long skuId = obj.getLong("sku_id");
                final Long ts = MyTimeUtil.toTs(obj.getString("create_time"));
                
                final ProductStats ps = new ProductStats();
                ps.setSku_id(skuId);
                ps.setTs(ts);
                ps.setComment_ct(1L);
                
                final String appraise = obj.getString("appraise");
                if (GmallConstant.APPRAISE_GOOD.equals(appraise)) {  // 硬编码  魔术数字
                    ps.setGood_comment_ct(1L);
                }
                
                return ps;
                
            });
        
        return productClickStream.union(
            displayStream,
            favorStream,
            cartStream,
            orderStream,
            paymentStream,
            refundStream,
            commentStream);
        
    }
    
}
